package cn.springboot.linked;
/**
 * @author liufl E-mail:Huberier@allpyra.com
 * @version 创建时间：2017年9月27日 下午4:35:47
 * 使用MapReduce及中文分词将爬到的文件制作成倒排索引，索引文件格式为  Key(分词器分出的词)+“\t”+url1:sum;url2:sum
 *  1.分布式爬虫：Nutch + hadoop
    2.JAVA单机爬虫：Crawler4j、WebMagic、WebCollector
 */
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.cn.smart.SmartChineseAnalyzer;
import org.apache.lucene.analysis.tokenattributes.CharTermAttribute;


//http://blog.csdn.net/m0_37739193/article/details/76572512   MapReduce操作
public class FindWord {

    public static class FindMapper extends Mapper<Text, Text, Text, Text>{



        @Override
        protected void map(Text key, Text value, Mapper<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            String text="android可行性";       //用户输入的关键字
            Analyzer sca = new SmartChineseAnalyzer( );  

            TokenStream ts = sca.tokenStream("field", text);  
            CharTermAttribute ch = ts.addAttribute(CharTermAttribute.class);  

            ts.reset();  
            while (ts.incrementToken()) {  
                if(ch.toString().equals(key.toString())||ch.toString().equals(key.toString())){
                    System.out.println(value.toString());
                    String[] urls=value.toString().split(";");
                    int count=0;
                    for (String url : urls) {
                        String oneurl=url.split(":")[0]+url.split(":")[1];
                        count=Integer.parseInt(url.split(":")[2]);
                        String newvalue=ch.toString()+";"+count;
                        System.out.println(">>>>>>>>"+oneurl+">>>>>>>>>>"+newvalue);
                        context.write(new Text(oneurl),new Text( newvalue));
                    }

                }
            }  
            ts.end();  
            ts.close();  



        }
    }

    public static class FindCombiner extends Reducer<Text, Text, Text, Text>{
        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {

            //统计词频
            int sum = 0;
            for (Text value : values) {
                String count=value.toString().split(";")[1];
                sum += Integer.parseInt(count );
            }
            context.write(new Text(String.valueOf(sum)),new Text(key.toString()) );
        }
    }


    public static class FindReducer extends Reducer<Text, Text, Text, Text>{

        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {

            //生成文档列表
            for (Text text : values) {
                context.write(key, text);
            }


        }

    }
    public static void main(String[] args) {


            try {
                Configuration conf = new Configuration();

                Job job = Job.getInstance(conf,"InvertedIndex");
                job.setJarByClass(InvertedIndex.class);

                //实现map函数，根据输入的<key,value>对生成中间结果。
                job.setMapperClass(FindMapper.class);

                job.setMapOutputKeyClass(Text.class);
                job.setMapOutputValueClass(Text.class);
                job.setInputFormatClass(KeyValueTextInputFormat.class);
                job.setCombinerClass(FindCombiner.class);
                job.setReducerClass(FindReducer.class);

                job.setOutputKeyClass(Text.class);
                job.setOutputValueClass(Text.class);


                FileInputFormat.addInputPath(job, new Path("hdfs://192.168.1.123:9000/spiderout/1462887403514/part-r-00000"));
                FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.1.123:9000/1"));

                System.exit(job.waitForCompletion(true) ? 0 : 1);
            } catch (IllegalStateException e) {
                e.printStackTrace();
            } catch (IllegalArgumentException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
}

